package nl.ViewerServer.ServerManager.JobManager.Strategy;

import java.io.File;
import java.io.PrintWriter;
import java.util.ArrayList;

import nl.ViewerServer.Application;
import nl.ViewerServer.ServerManager.ServerManager;
import nl.ViewerServer.ServerManager.JobManager.LocalityThreadAssignment;
import nl.ViewerServer.ServerManager.JobManager.ParaViewMachine;
import nl.ViewerServer.ServerManager.JobManager.ParaViewThreadAssignment;

import org.apache.log4j.Logger;

/**
 * This strategy will try to assign the requested threads evenly among the physical processors on single machines. This
 * strategy will adhere to the set maximum amount of ParaView threads (+ /per worker) in the config. The strategy has
 * the following properties: 1) The strategy will pick the nodes that have either the most physical threads available,
 * nodes that have more than the maximum allowed threads (in the config) available, or nodes that have more than the
 * requested number of threads available. 2) If equal (or equal because of the defined config) keep the
 * 'overbook-factor' (ratio between occupied threads and physical cores) as low as possible. 3) If equal choose the host
 * with the highest rating (which has faster processor cores for example).
 * 
 * @author Jeroen Verschuur
 * 
 */

public class LocalityStrategy implements AssignmentStrategy {
   private static Logger                         log               = Logger.getLogger( LocalityStrategy.class );

   protected ArrayList<ParaViewThreadAssignment> activeAssignments = new ArrayList<ParaViewThreadAssignment>();
   protected ArrayList<ParaViewMachine>          paraViewMachines;
   protected int                                 threadsAssigned   = 0;

   public static enum Decision {
      MOST_THREADS_AVAILABLE, LOWEST_OVERBOOK, HIGHEST_RATING, UNDECIDED
   };

   public LocalityStrategy() {
      // get some info from the config

      // deep-copy the paraview nodes, we will 'abuse' the ParaViewNode object, and use the coreCount member as an
      // 'available (physical) core counter'
      paraViewMachines = new ArrayList<ParaViewMachine>( Application.config.paraViewMachines.size() );
      for( ParaViewMachine node : Application.config.paraViewMachines ) {
         paraViewMachines.add( new ParaViewMachine( node.host, node.coreCount, node.rating ) );
      }
   }

   /**
    * This method will check the current amount of active workers, the maximum amount of threads per job and the maximum
    * total amount of threads assignable and calculate the maximum number of threads that can be assigned.
    * 
    * @return
    */
   public int getCurrentMaximumAssignableThreads() {
      // calculate the amount of threads to be assigned at most (the minimum of the max per thread count and the amount
      // of unemployed worker threads)
      int workersLeft = Application.config.poolSize - ServerManager.jobManager.getActiveCount();
      int currentMaximum = Math.min( Application.config.paraviewThreadCount - threadsAssigned - workersLeft,
            Application.config.maxParaviewThreadCountPerWorker );

      return currentMaximum;
   }

   /**
    * Select those nodes that can run most (physical) threads in parallel
    * 
    * @param nodes
    * @return
    */
   public ArrayList<ParaViewMachine> getAvailableCandidates( ArrayList<ParaViewMachine> nodes, int maximumAssignable ) {
      ArrayList<ParaViewMachine> mostAvailable = new ArrayList<ParaViewMachine>( 1 );
      mostAvailable.add( paraViewMachines.get( 0 ) );

      int currentNumberOfThreads = getNumberOfThreads( mostAvailable.get( 0 ), maximumAssignable );

      // check which server has (or which servers have) most threads available
      for( int i = 1; i < paraViewMachines.size(); i++ ) {
         ParaViewMachine node = paraViewMachines.get( i );
         int newNumberOfThreads = getNumberOfThreads( node, maximumAssignable );
         if( newNumberOfThreads >= currentNumberOfThreads ) {
            if( newNumberOfThreads > currentNumberOfThreads ) {
               mostAvailable.clear();
               currentNumberOfThreads = newNumberOfThreads;
            }

            mostAvailable.add( node );
         }
      }

      return mostAvailable;
   }

   private static int getNumberOfThreads( ParaViewMachine node, int maximumAssignable ) {
      return node.coreCount > 0 ? Math.min( maximumAssignable, node.coreCount ) : 1;
   }

   public ParaViewThreadAssignment assignParaViewThreads( int requestedThreads ) {
      if( paraViewMachines.size() == 0 )
         return null;

      int maximumAssignable;

      if( requestedThreads > 0 )
         maximumAssignable = Math.min( getCurrentMaximumAssignableThreads(), requestedThreads );
      else
         maximumAssignable = getCurrentMaximumAssignableThreads();

      // this should never happen (indicates a bug or wrong configuration)!
      if( maximumAssignable < 0 ) {
         log.error( "MaximumAssignable threads for assignParaview is not allowed: " + maximumAssignable );
         maximumAssignable = 1;
      }

      Decision decision = Decision.UNDECIDED;

      ArrayList<ParaViewMachine> candidates = getAvailableCandidates( paraViewMachines, maximumAssignable );

      // log.info( "# available: " + candidates.size() );

      // at this point we know how many threads will be appointed, this number will be used in the overbook calculation
      // below
      int numberOfThreads = candidates.size() > 0 ? getNumberOfThreads( candidates.get( 0 ), maximumAssignable ) : -1;

      if( candidates.size() > 1 ) {
         // We should narrow down the choice: choose the candidate with the lowest 'overbook-factor'. If there are nodes
         // with the same overbook-factor, use the one with the highest rating.
         ParaViewMachine currentLowest = null;
         double lowestOverbook = Double.MAX_VALUE;

         double overbook;
         ParaViewMachine c;
         for( int i = 0; i < candidates.size(); i++ ) {
            c = candidates.get( i );
            // We actually want to compare the situation that will arise AFTER assigning the threads to the new machine.
            // So, in the overbook calculation below we will account for the amount of threads we will be handing out.
            // Example: you have two servers with 4 and 2 cores respectively. The first has two
            // threads in use, the second has 1 thread in use. We need to assign a single thread to a new client
            // session. The current overbooking for the servers is equal, but the new overbooking situation would be
            // 3/4th, 2/2nd respectively, so we should favor the quad-core server in this case (before looking at the
            // processor rating).
            //
            overbook = (double)( c.physicalCoreCount - c.coreCount + numberOfThreads ) / (double)c.physicalCoreCount;
            // log.info( "test candidate: " + c + ", overbook: " + overbook );
            // choose the one with lowest overbook, or, if equal, the one with highest rating
            if( currentLowest == null || overbook < lowestOverbook || ( overbook == lowestOverbook && c.rating > currentLowest.rating ) ) {
               // log.info( "New better candidate" );
               if( overbook < lowestOverbook ) {
                  decision = Decision.LOWEST_OVERBOOK;
               } else {
                  decision = Decision.HIGHEST_RATING;
               }
               // log.info( "decision: " + decision.toString() );
               currentLowest = c;
               lowestOverbook = overbook;
            } else {
               // if the candidate-to-test is not better, but the overbook is the same as the current candidate, then
               // the earlier decision for a candidate was actually based on the rating instead of the overbook
               if( overbook == lowestOverbook ) {
                  if( c.rating == currentLowest.rating )
                     decision = Decision.UNDECIDED;
                  else
                     decision = Decision.HIGHEST_RATING;
               }
            }
         }
         candidates.clear();
         candidates.add( currentLowest );

      } else if( candidates.size() == 1 ) {
         decision = Decision.MOST_THREADS_AVAILABLE;
      }

      if( candidates.size() == 0 ) {
         log.warn( "Could not assign threads to the worker" );
         return null;
      }

      // update the information in paraViewNodes
      ParaViewMachine candidate = candidates.get( 0 );
      // Choose the amount of cores physically available, or otherwise 1. BTW, this choice of assigning only a single
      // thread when there are no more physical threads available is debatable.
      candidate.coreCount -= numberOfThreads;

      log.info( "Decision: " + decision.toString() + ": assigned to host " + candidates.get( 0 ).host + ", " + numberOfThreads + " threads (new coreCount for host: " + candidate.coreCount + ")" );

      LocalityThreadAssignment assignment = new LocalityThreadAssignment( numberOfThreads,
            generateSingleHostMachineFile( candidate.host, numberOfThreads ), candidates, new int[] { numberOfThreads } );

      threadsAssigned += numberOfThreads;

      return assignment;
   }

   public void reclaimParaviewThreadAssignment( ParaViewThreadAssignment assignment ) {
      if( assignment == null ) {
         log.warn( "Cannot reclaim null-assignment" );
         return;
      }

      // the LocalityStrategy will always provide an assignment of type LocalityThreadAssignment
      if( assignment instanceof LocalityThreadAssignment == false ) {
         log.error( "Cannot reclaim paraview assignments which are not assigned by a LocalityStrategy" );
         return;
      }

      // update the information in paraViewNodes
      LocalityThreadAssignment lta = (LocalityThreadAssignment)assignment;

      for( int i = 0; i < lta.nodesAssigned.size(); i++ ) {
         lta.nodesAssigned.get( i ).coreCount += lta.threadsAssignedForNodes[i];
         log.info( "Reclaimed PV thread for " + lta.nodesAssigned.get( i ) + ", " + lta.threadsAssignedForNodes[i] + " threads" );
      }

      threadsAssigned -= lta.totalThreadCount;

   }

   public String generateSingleHostMachineFile( String host, int numberOfThreads ) {

      try {
         new File( Application.config.dataStore + "/machinefiles" ).mkdirs();

         File tempFile = new File( Application.config.dataStore + "/machinefiles/",
               host + "_" + numberOfThreads + ".hosts" );

         if( !tempFile.exists() ) {
            PrintWriter pw = new PrintWriter( tempFile );
            pw.println( host + ":" + numberOfThreads );
            pw.close();
         }

         return tempFile.getCanonicalPath();
      }
      catch( Exception e ) {
         // in any exception, default back to the mpiMachinefile defined in the config
         log.warn( "Could not generate machinefile needed for the chosen strategy: ", e );
         return Application.config.MPIMachineFile;
      }
   }
}
